package mr.stocks.question3;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import mr.stocks.avro.model.StockPriceByYearKeyAvro;
import mr.stocks.avro.model.StockPriceByYearReportAvro;
import mr.stocks.avro.model.StockPriceByYearValueAvro;
import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.stocks.avro.model.StocksValuesAvro;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class StockPriceByYearAvroMapper extends
		Mapper<AvroKey<StocksReportAvro>, NullWritable, Text, AvroValue<StockPriceByYearReportAvro>> {
	
	private static Logger LOG = LoggerFactory.getLogger(StockPriceByYearAvroMapper.class);

	private StockPriceByYearReportAvro outValue = new StockPriceByYearReportAvro();
	private Map<CharSequence, StockPriceByYearValueAvro> priceMap = new HashMap<CharSequence, StockPriceByYearValueAvro>();
	private AvroValue<StockPriceByYearReportAvro> outValueArvo = new AvroValue<StockPriceByYearReportAvro>();
	
	StockPriceByYearValueAvro stockPrice = new StockPriceByYearValueAvro();
	private String yearStr;
	private String currentStock;

	private Text keyText = new Text();

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
	}

	@Override
	protected void map(AvroKey<StocksReportAvro> inKey, NullWritable inValue,
			Context context) throws IOException, InterruptedException {

		StocksKeyAvro date = inKey.datum().getYearMonthDay();
		yearStr = date.getYear().toString();

		List<StocksValuesAvro> stockList = inKey.datum().getStocks();

		for (StocksValuesAvro stock : stockList) {
			currentStock = stock.getStock().toString();

			stockPrice.setCount((long)1);
			stockPrice.setPrice(stock.getPrice());
			priceMap.put(yearStr, stockPrice);
			
			outValue.setAverageAnnualPrice(priceMap);
			outValue.setStockSymbol(currentStock);
			outValueArvo.datum(outValue);
			
			keyText.set(currentStock);
			context.write(keyText, outValueArvo);
		}
	}
}
